package com.zhang.third.day07;

import com.zhang.third.utils.ClickEventSource;
import com.zhang.third.utils.Event;
import com.zhang.third.utils.UserViewCountPerWindow;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * @title: 触发器应用--统计一天pv，但是当窗口第一条数据来之后每隔一秒输出统计信息不销毁窗口
 * @author: zhang
 * @date: 2022/4/11 14:38
 */
public class Example4 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new ClickEventSource())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Event>forMonotonousTimestamps()
                                .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                                    @Override
                                    public long extractTimestamp(Event element, long recordTimestamp) {
                                        return element.ts;
                                    }
                                })
                )
                .keyBy(r -> r.key)
                .window(TumblingEventTimeWindows.of(Time.days(1)))
                .trigger(new Trigger<Event, TimeWindow>() {
                    // 每来一条数据执行一次
                    @Override
                    public TriggerResult onElement(Event element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
                        ValueState<Boolean> flag = ctx.getPartitionedState(
                                new ValueStateDescriptor<Boolean>(
                                        "flag", Types.BOOLEAN
                                )
                        );
                        // 只有当属于窗口的第一条数据到来才执行if逻辑
                        if (flag.value() == null) {
                            // 将标志位置为true，那么第二条以及第二条之后的数据不会进入到if语句中
                            flag.update(true);
                            // 注册属于窗口第一条数据之后的下一个整数秒
                            long nextSecond = element.ts + 1000L - element.ts % 1000L;
                            ctx.registerEventTimeTimer(nextSecond);
                        }
                        // 当onElement调用完时，不对窗口做任何事情
                        return TriggerResult.CONTINUE;
                    }

                    @Override
                    public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
                        return null;
                    }

                    // 到达trigger的水位线超过参数`time`时，触发调用
                    @Override
                    public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
                        if (time < window.getEnd()) {
                            if (time + 1000L < window.getEnd()) {
                                ctx.registerEventTimeTimer(time + 1000L);
                            }
                            // 触发后面`aggregate`逻辑计算但是不销毁窗口
                            return TriggerResult.FIRE;
                        }
                        // 如果注册时间大于窗口时间什么都不做
                        return TriggerResult.CONTINUE;
                    }

                    @Override
                    public void clear(TimeWindow window, TriggerContext ctx) throws Exception {
                        // 单例对象
                        ValueState<Boolean> flag = ctx.getPartitionedState(
                                new ValueStateDescriptor<Boolean>(
                                        "flag", Types.BOOLEAN
                                )
                        );
                        flag.clear();
                    }
                })
                .aggregate(
                        new AggregateFunction<Event, Long, Long>() {
                            @Override
                            public Long createAccumulator() {
                                return 0L;
                            }

                            @Override
                            public Long add(Event value, Long accumulator) {
                                return accumulator + 1L;
                            }

                            @Override
                            public Long getResult(Long accumulator) {
                                return accumulator;
                            }

                            @Override
                            public Long merge(Long a, Long b) {
                                return null;
                            }
                        },
                        new ProcessWindowFunction<Long, UserViewCountPerWindow, String, TimeWindow>() {
                            @Override
                            public void process(String s, ProcessWindowFunction<Long, UserViewCountPerWindow, String, TimeWindow>.Context context, Iterable<Long> elements, Collector<UserViewCountPerWindow> out) throws Exception {
                                out.collect(new UserViewCountPerWindow(
                                        s,
                                        elements.iterator().next(),
                                        context.window().getStart(),
                                        context.window().getEnd()
                                ));
                            }
                        }

                )
                .print();

        env.execute();
    }
}
